package com.atgugu.flink.chapter07.state;

import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @Author lzc
 * @Date 2022/4/6 14:19
 */
public class Flink03_Operator_BroadCast {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);
        env.enableCheckpointing(3000);  // 每3s把状态做做一次快照
        
        // 广播状态使用的套路
        /*
        1. 需要两个流,
            一个流是数据类,
            另外一个流是控制流,用去控制数据流中每条数据的处理方式
        
        2. 把控制流做成一个广播流. 将来会把广播流中的数据广播到数据流中, 广播出去的数据就是广播状态
        
        3. 用数据流去connect广播流
        
        4. 对connect之后的进行处理
            处理广播流中的数据
                把数据存入到广播状态中
                
            
            处理数据流中的数据
                从广播状态中获取广播数据
       
         */
        
        DataStreamSource<String> dataStream = env
            .socketTextStream("hadoop162", 8888);
        DataStreamSource<String> controlStream = env
            .socketTextStream("hadoop162", 9999);
        
        MapStateDescriptor<String, String> bcStateDesc = new MapStateDescriptor<>("bcState", String.class, String.class);
        
        // controlStream 做成广播流
        BroadcastStream<String> bcStream = controlStream.broadcast(bcStateDesc);
        
        BroadcastConnectedStream<String, String> bcConnectedStream = dataStream.connect(bcStream);
        
        bcConnectedStream
            .process(new BroadcastProcessFunction<String, String, String>() {
                // 处理数据流中的元素: 从广播状态读取数据, 然后根据读到的配置信息, 来处理数据
                @Override
                public void processElement(String value,
                                           ReadOnlyContext ctx,
                                           Collector<String> out) throws Exception {
                    System.out.println("Flink03_Operator_BroadCast.processElement");
                    ReadOnlyBroadcastState<String, String> bcState = ctx.getBroadcastState(bcStateDesc);
    
                    if (value.contains("sensor_1")) {
                        out.collect("sensor_1 采用 :" + (bcState.get("sensor_1")==null ? "default" :bcState.get("sensor_1"))  + "号处理逻辑");
                    } else if (value.contains("sensor_2")) {
                        out.collect("sensor_2 采用 :" + (bcState.get("sensor_2")==null ? "default" :bcState.get("sensor_2")) + "号处理逻辑");
                    }
                }
                
                // 处理广播流中的元素: 把广播流中的元素存入到广播状态
                @Override
                public void processBroadcastElement(String value,
                                                    Context ctx,
                                                    Collector<String> out) throws Exception {
                    System.out.println("Flink03_Operator_BroadCast.processBroadcastElement");
                    // sensor_1, 1
                    // sensor_1, 2
                    // sensor_1, 3
                    // sensor_2, 1
                    // sensor_3, 1
                    String[] data = value.split(",");
                    BroadcastState<String, String> broadcastState = ctx.getBroadcastState(bcStateDesc);
                    
                    broadcastState.put(data[0], data[1]);
                    
                    
                }
            })
            .print();
        
        
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    
}
